package cn.itcast.c_watermark;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

/**
 * 目标:
 * 掌握水印/水位线的设置和使用
 */
public class WarterMarkDemo {
    //需求: 以EventTime划分窗口，计算每3秒钟内出价最高的产品

    /**
     * 分析需求：
     * 1.使用滚动时间窗口，窗口大小 = 3s
     * 2.求每3s内，价格最高的产品
     */

    public static void main(String[] args) throws Exception {

        /**
         * 开发步骤：
         * 1.执行环境
         * 2.加载数据源
         * 3.数据转换：新建bean对象
         * 4.设置水位线
         * 5.数据分组
         * 6.划分时间窗口
         * 7.数据聚合
         * 8.打印
         * 9.触发执行
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);//单并行度演示
        DataStreamSource<String> source = env.socketTextStream("node1", 8090);
        //3.数据转换：新建bean对象
        //1633146355000,tesla_2,model_3,100.0
        SingleOutputStreamOperator<ModelCar> mapData = source.map(new MapFunction<String, ModelCar>() {
            @Override
            public ModelCar map(String value) throws Exception {
                String[] arr = value.split(",");
                return new ModelCar(
                        Long.valueOf(arr[0]),
                        arr[1],
                        arr[2],
                        Double.valueOf(arr[3])
                );
            }
        });

        // 4.设置水位线
        SingleOutputStreamOperator<ModelCar> waterData
                = mapData.assignTimestampsAndWatermarks(WatermarkStrategy.<ModelCar>forBoundedOutOfOrderness(Duration.ofSeconds(2))//延迟2s
                .withTimestampAssigner(new SerializableTimestampAssigner<ModelCar>() {
                    @Override
                    public long extractTimestamp(ModelCar element, long recordTimestamp) {
                        return element.getEventTime();
                    }
                }));
        //5.数据分组
        waterData.keyBy(new KeySelector<ModelCar, String>() {
                    @Override
                    public String getKey(ModelCar value) throws Exception {
                        return value.getCompany();
                    }
                })
                //6.划分时间窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(4))) //设置窗口大小
                //7.数据聚合
                .max("price")
                //8.打印
                .print();

        //9.触发执行
        env.execute();
    }
}
